package com.atguigu.flink.chapter08_exec2;

import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.MyUtil;
import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/10/29
 *
 *      假设规定，30min之内必须完成订单的支付，否则就是失效。
 *          希望找出有问题的数据。
 *
 *      支付的场景有以下情况:
 *          ①正常:  同一笔订单，有create,有pay，时间间隔在30min之内
 *          ④不正常:  同一笔订单，有create,有pay，时间间隔超过30min  超时
 *          ②正常:  同一笔订单，只有create，没有支付
 *          ③不正常:  同一笔订单，只有pay，没有create
 *
 *      ---------------------------
 *        业务场景：  先提交订单，再支付订单。
 *                  同一笔订单，先有create,后有pay
 *
 *           找不正常:
 *                 根据orderId分组，分组后开启一个 SessionWindow(30min)。
 *                 当window关闭计算时，判断看窗口中是否只有1个数据。
 *                  如果只有一个数据，判断是create还是pay,如果是create，写入状态保存。
 *                      如果后续有pay到达，会自己再开个窗口，窗口关闭时，窗口就它自己。
 *                          触发计算，让pay去状态中找create，找到了，说明超时支付，
 *                              找不到，基于业务限定，认为缺少create
 *
 */
public class Demo6_Order
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
            .withTimestampAssigner((u, ts) -> u.getEventTime());



        env.readTextFile("data/OrderLog.csv").setParallelism(1)
           .map(new MapFunction<String, OrderEvent>()
           {
               @Override
               public OrderEvent map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new OrderEvent(
                       Long.parseLong(words[0]),
                       words[1],
                       words[2],
                       Long.parseLong(words[3]) * 1000
                   );
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           .keyBy(OrderEvent::getOrderId)
           .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
           .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>()
           {

               private ValueState<OrderEvent> create;

               @Override
               public void open(Configuration parameters) throws Exception {
                   create = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("create", OrderEvent.class));
               }

               @Override
               public void process(Long orderId, Context context, Iterable<OrderEvent> elements, Collector<String> out) throws Exception {

                   List<OrderEvent> orderEvents = MyUtil.toList(elements);

                   if (orderEvents.size() == 1 ){
                       //异常情况
                       OrderEvent orderEvent = orderEvents.get(0);

                       //两种情况，一种是超时，一种是缺失create
                       if ("pay".equals(orderEvent.getEventType()) ){

                           if (create.value() != null){
                               out.collect(orderId + "超时支付...");
                           }else{
                               out.collect(orderId+"缺少create信息....");
                           }
                       }else {
                           //说明当前的数据是create，存入状态
                           create.update(orderEvent);
                       }
                   }

               }
           }).print().setParallelism(1);

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
